package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.bean.TradeOrderBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.DorisUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * ClassName: DwsTradeOrderWindow
 * Package: com.atguigu.edu.realtime.app.dws
 * Description:
 * 交易域下单独立用户数和新增下单用户数
 * @Author Mr.2
 * @Create 2023/9/11 16:56
 * @Version 1.0
 */

public class DwsTradeOrderWindow {
    public static void main(String[] args) {
        // TODO 1. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置 检查点超时时间
        // 2.3 设置 job取消后，检查点是否保留
        // 2.4 设置 检查点重启策略 -- 可以设置故障率重启策略
        // 2.5 设置 两个检查点之间最小时间间隔
        // 2.6 设置 状态后端(即: 检查点保存位置)
        // 2.7 设置 操作系统的用户

        // TODO 3. From kafka topic 读取数据
        // 3.1 声明 消费者主题 和 消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_order_window_group_01";
        // 3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        // 3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrStream =
                env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");

        // TODO 4. 过滤 null, 对数据 进行数据类型转换 JSONString -> JSONObject
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStrStream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        // 过滤 空消息
                        if (jsonStr != null) {
                            JSONObject jsonObject = JSON.parseObject(jsonStr);
                            // 向下游发送
                            out.collect(jsonObject);
                        }
                    }
                }
        );

        // TODO 5. 指定watermark, 并提取事件时间字段
        /* 流中的数据样式:
        * {
        *      "id": "26182",
        *      "order_id": "25024",
        *      "user_id": "695",
        *      "course_id": "403",
        *      "course_name": "大数据项目之电商数仓2.0",
        *      "province_id": "31",
        *      "session_id": "014b3288-907f-4f58-9714-6a0f7dfb0430",
        *      "date_id": "2023-09-08",
        *      "create_time": "2023-09-08 19:28:12",
        *      "origin_amount": "200.0",
        *      "coupon_reduce": "0.0",
        *      "final_amount": "200.0",
        *      "ts": "1694172492"  -- 特别注意: 单位为s
        *    }
        *  Note that: 指定watermark时的时间戳，单位必须为ms
        * */
        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonObjStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        // 秒 -> 毫秒
                                        return element.getLong("ts") * 1000;
                                    }
                                }
                        )
        );

        // TODO 6. 按照用户id进行分组
        KeyedStream<JSONObject, String> keyedByUserIdStream =
                withWatermarkStream.keyBy(jsonObj -> jsonObj.getString("user_id"));

        // TODO 7. 统计当日下单独立用户数 和 新增下单用户数
        SingleOutputStreamOperator<TradeOrderBean> processedStream = keyedByUserIdStream.process(
                new KeyedProcessFunction<String, JSONObject, TradeOrderBean>() {

                    // 声明状态; 存放各用户的上个订单日期
                    private ValueState<String> lastOrderDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //
                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<>("lastOrderDateState", String.class);
                        // 思考: 需要不需要设置 状态保留时间TTL

                        lastOrderDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<TradeOrderBean> out) throws Exception {
                        // 获取 用户上次下单日期
                        String lastOrderDate = lastOrderDateState.value();

                        // 初始化 统计指标
                        long orderNewUserCount = 0L;
                        long orderUniqueUserCount = 0L;
                        // !!! 注意时间戳
                        // 获取当前订单日期
                        long ts = jsonObj.getLong("ts") * 1000L;
                        String curOrderDate = DateFormatUtil.toDate(ts);

                        if (lastOrderDate == null) {
                            orderNewUserCount = 1L;
                            orderUniqueUserCount = 1L;
                        } else {
                            // 上次订单日期不为空 且 上次订单日期不等于当前订单日期
                            if (!lastOrderDate.equals(curOrderDate)) {
                                orderUniqueUserCount = 1L;
                                lastOrderDateState.update(curOrderDate);
                            }
                        }

                        TradeOrderBean orderBean =
                                new TradeOrderBean("", "", "", orderUniqueUserCount, orderNewUserCount);

                        out.collect(orderBean);
                    }
                }
        );

        // TODO 8. 开窗
        AllWindowedStream<TradeOrderBean, TimeWindow> windowalledStream
                = processedStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10L)));

        // TODO 9. 聚合
        SingleOutputStreamOperator<TradeOrderBean> reducedStream = windowalledStream.reduce(
                new ReduceFunction<TradeOrderBean>() {
                    @Override
                    public TradeOrderBean reduce(TradeOrderBean value1, TradeOrderBean value2) throws Exception {
                        value1.setOrderNewUserCount(value1.getOrderNewUserCount() + value2.getOrderNewUserCount());
                        value1.setOrderUniqueUserCount(value1.getOrderUniqueUserCount() + value2.getOrderUniqueUserCount());
                        return value1;
                    }
                },
                // 对聚合后的数据, 补充时间相关属性(窗口开启、窗口关闭、Doris分区字段)
                new ProcessAllWindowFunction<TradeOrderBean, TradeOrderBean, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<TradeOrderBean> elements, Collector<TradeOrderBean> out) throws Exception {

                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String curDate = DateFormatUtil.toDate(context.window().getStart());

                        for (TradeOrderBean bean : elements) {
                            bean.setStt(stt);
                            bean.setEdt(edt);
                            bean.setCurDate(curDate);

                            //
                            out.collect(bean);
                        }
                    }
                }
        );
        // For test output
        reducedStream.print("reduced->");

        // TODO 10. 写到Doris
        reducedStream
                .map(new BeanToJsonStrFunction<TradeOrderBean>())
                .sinkTo(DorisUtil.getDorisSink("dws_trade_order_window"));

        // 执行 环境
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
